package com.rpc.example.zookeeper;

import com.rpc.example.IRegitserService;
import com.rpc.example.ServiceInfo;
import com.rpc.example.loadbalance.ILoadBalance;
import com.rpc.example.loadbalance.RandomLoadBanlance;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.x.discovery.ServiceDiscovery;
import org.apache.curator.x.discovery.ServiceDiscoveryBuilder;
import org.apache.curator.x.discovery.ServiceInstance;
import org.apache.curator.x.discovery.details.JsonInstanceSerializer;

import java.util.Collection;
import java.util.List;

/**
 *
 *
 * @author Elvis
 * @version 1.0, 2022/11/21
 */
public class ZookeeperRegisterService implements IRegitserService {
    private static final String REGISTER_PATH = "/register";

    private final ServiceDiscovery<ServiceInfo> serviceDiscovery;

    private ILoadBalance<ServiceInstance<ServiceInfo>> loadBalance;

    public ZookeeperRegisterService(String registryAddress) throws Exception {
        CuratorFramework client = CuratorFrameworkFactory
                .newClient(registryAddress, new ExponentialBackoffRetry(1000, 3));
        client.start();
        ;
        JsonInstanceSerializer<ServiceInfo> serviceInfo = new JsonInstanceSerializer<>(ServiceInfo.class);
        this.serviceDiscovery = ServiceDiscoveryBuilder.builder(ServiceInfo.class)
                .client(client)
                .serializer(serviceInfo)
                .basePath(REGISTER_PATH)
                .build();
        this.serviceDiscovery.start();
        loadBalance = new RandomLoadBanlance();
    }

    @Override
    public void register(ServiceInfo serviceInfo) throws Exception {
        ServiceInstance<ServiceInfo> serviceInstance = ServiceInstance.<ServiceInfo>builder()
                .name(serviceInfo.getServiceName())
                .address(serviceInfo.getServiceAddress())
                .port(serviceInfo.getServicePort())
                .payload(serviceInfo)
                .build();
        serviceDiscovery.registerService(serviceInstance);
    }

    @Override
    public ServiceInfo discovery(String serviceName) throws Exception {
        Collection<ServiceInstance<ServiceInfo>> serviceInstances = this.serviceDiscovery.queryForInstances(serviceName);
        return loadBalance.select((List<ServiceInstance<ServiceInfo>>) serviceInstances).getPayload();
    }
}
